package com.WaVel.RabbitMqComponent;

import com.Emplyment.dto.QuestionBankDto;
import com.Emplyment.entity.Article;
import com.Emplyment.entity.QuestionBank;
import com.Emplyment.exception.BaseException;
import com.Emplyment.exception.RegisterException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Component
@Slf4j
public class consumer {

    @Autowired
    private RestHighLevelClient client;

    // 获取那个消息队列    消息队列的名称
//    @RabbitListener(queues = "simple.queue")
//    public  void  listenSimpleQueueMessage(String message) {
//        System.out.println("spring 消费者接收到的消息是："+message);
//    }

    /**
     * 创建公司时会执行
     * @param msg
     */
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg){

        Article article = JSON.parseObject(msg, Article.class);
        IndexRequest request = new IndexRequest("app_emplyment_message").id(article.getId().toString());
        request.source(msg, XContentType.JSON);

        try {
            client.index(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RegisterException("消费者1添加到索引库错误");
        }
        System.out.println("消费者1添加公司数据接收到Fanout消息：【" + msg + "】");
    }

    /**
     * 更新公司时会执行
     * @param msg
     */
    @RabbitListener(queues = "fanout.queue2")
    public void updateFanoutQueue(String msg){

        Article article = JSON.parseObject(msg, Article.class);
        UpdateRequest request = new UpdateRequest("app_emplyment_message", article.getId().toString());
        request.doc(msg, XContentType.JSON);

        try {
            client.update(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RegisterException("消费者2修改索引库错误");
        }
        System.out.println("消费者2更新公司数据接收到Fanout消息：【" + msg + "】");
    }

    /**
     * 删除公司数据时会执行
     * @param id
     */
    @RabbitListener(queues = "fanout.queue3")
    public void deleteFanoutQueue(String id){
        System.out.println("消费者3删除公司数据接收到Fanout消息：【" + id + "】");
        try {
            // 1.准备Request
            DeleteRequest request = new DeleteRequest("app_emplyment_message", id);
            // 2.发送请求
            client.delete(request, RequestOptions.DEFAULT);
        }catch (IOException e){
            throw new BaseException("消费者3删除索引库错误");
        }
    }


    /**
     * 添加面试题库时会执行
     * @param msg
     */
    @RabbitListener(queues = "interview.add.queue")
    public void listenFanoutQueue2(String msg){
        System.out.println("消费者4添加面试题接收到Fanout消息：【" + msg + "】");
        QuestionBank questionBank = JSON.parseObject(msg, QuestionBank.class);
        IndexRequest request = new IndexRequest("interview_emplyment_message").id(questionBank.getId().toString());
        request.source(msg, XContentType.JSON);
        try {
            client.index(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RegisterException("添加到索引库错误");
        }
    }

    /**
     * 更新面试题库时会执行
     * @param msg
     */
    @RabbitListener(queues = "interview.update.queue")
    public void updateInterviewQueue(String msg){
        log.info("发送的消息为:{}",msg);
        QuestionBank questionBank = JSON.parseObject(msg, QuestionBank.class);
        Map<String, Object> doc = JSON.parseObject(msg, new TypeReference<Map<String, Object>>() {});
        UpdateRequest request = new UpdateRequest("interview_emplyment_message", questionBank.getId().toString());

        // 将消息解析为 Map，用于更新文档

        request.doc(doc, XContentType.JSON);

        // 启用 upsert：如果文档不存在，则插入
        request.docAsUpsert(true);

        try {
            client.update(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RegisterException("消费者更新面试问题后修改索引库错误");
        }
        System.out.println("消费者更新面试问题数据接收到Fanout消息：【" + msg + "】");
    }

    /**
     * 删除面试题库时会执行
     * @param id
     */
    @RabbitListener(queues = "interview.delete.queue")
    public void deleteInterviewQueue(String id){
        System.out.println("消费者6删除面试数据接收到Fanout消息：【" + id + "】");
        try {
            // 1.准备Request
            DeleteRequest request = new DeleteRequest("interview_emplyment_message", id);
            // 2.发送请求
            client.delete(request, RequestOptions.DEFAULT);
        }catch (IOException e){
            throw new BaseException("消费者6删除索引库错误");
        }
    }

}
